package cn.xiaoniu.dmp.report


import java.util

import cn.xiaoniu.dmp.helper.ConfigHelper
import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}
import utils.JedisPoolUtils

object MediaAnalysis {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("媒体分析")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val file: RDD[String] = sc.textFile("D:\\app_dict.txt")
    val frame: DataFrame = sqlContext.read.parquet("D:/dmp/parquet")

    val map: Map[String, String] = file.filter(_.split("\\\t",-1).length>=2).map(t => {
      val sp: Array[String] = t.split("\\\t",-1)
      (sp(0), sp(1))
    }).collect().toMap//toMAp的时候会有相同的key进行覆盖
    println(map.size)

    //方式1 hmset  scala 转java map 保存redis中
//    import collection.JavaConversions._
//    val  mapp:util.Map[String,String]=map
//          val jedis = JedisPoolUtils.getJedis()
//    jedis.hmset("map",mapp)
//    jedis.close()

   //方式二 hset for循环保存reids中
    val jedis = JedisPoolUtils.getJedis()
              map.foreach(t=>{
                jedis.hset("reference",t._1,t._2)
              })
    jedis.close()

     // 广播变量方式
    //    val bd: Broadcast[Map[String, String]] = sc.broadcast(map)
//    sqlContext.udf.register("findName",(appname:String,appid:String)=> {
//      if (StringUtils.isNotBlank(appname)) appname else if (StringUtils.isNotBlank(appid)) bd.value.get(appid).getOrElse("unknow") else "unknow"
//    } )


    // jedis hmset方式
//    sqlContext.udf.register("findName",(appname:String,appid:String)=> {
//      val jedis2 = JedisPoolUtils.getJedis()
//      val result:String = if (StringUtils.isNotBlank(appname)) appname else if (StringUtils.isNotBlank(appid)) jedis2.hmget("map",appid).get(0) else "unknow"
//         jedis2.close()
//         result
//        } )

    //jedis  hset方式
    sqlContext.udf.register("findName",(appname:String,appid:String)=> {
      val jedis2 = JedisPoolUtils.getJedis()
      val result:String = if (StringUtils.isNotBlank(appname)) appname else if (StringUtils.isNotBlank(appid)) jedis2.hget("reference",appid) else "unknow"
      jedis2.close()
      result
    } )

    frame.registerTempTable("t_media")
    sqlContext.sql("select findName(appname,appid)appname,requestmode,processnode,iseffective,isbilling,isbid," +
      "iswin,adorderid,winprice,adpayment from t_media").registerTempTable("t_m")

    val result: DataFrame = sqlContext.sql(
      """
      select appname,
       sum(if (requestmode=1 and processnode>=2,1,0 )) effectiveReq,
       sum(if (requestmode=1 and processnode=3,1,0 )) adReq,

       sum(if (iseffective=1 and isbilling=1 and isbid=1 and adorderid!=0,1,0 )) bid,
       sum(if (iseffective=1 and isbilling=1 and iswin=1,1,0 )) succbid,

       sum(if (requestmode=2 and iseffective=1,1,0 )) display,
       sum(if (requestmode=3 and iseffective=1,1,0 )) click,

      sum(if (iseffective=1 and isbilling=1  and iswin=1,winprice/1000,0 )) cost,
      sum(if (iseffective=1 and isbilling=1 and iswin=1,adpayment/1000,0 )) spend
      from t_m group by appname
      """.stripMargin)

    result.write.jdbc(ConfigHelper.url,ConfigHelper.table,ConfigHelper.props)
    sc.stop()
  }
}
